package com.jinyang.jetpackdemo.activity

import androidx.appcompat.app.AppCompatActivity
import android.os.Bundle
import com.jinyang.jetpackdemo.R
import com.jinyang.jetpackdemo.util.LjyLogUtil
import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers
import io.reactivex.rxjava3.core.*
import io.reactivex.rxjava3.disposables.CompositeDisposable
import io.reactivex.rxjava3.disposables.Disposable
import io.reactivex.rxjava3.schedulers.Schedulers
import org.reactivestreams.Subscriber
import org.reactivestreams.Subscription
import java.io.Serializable
import java.util.concurrent.TimeUnit
import android.R.string.no
import java.lang.Exception
import android.R.string.no
import android.widget.Button
import android.widget.EditText
import com.google.gson.annotations.SerializedName
import com.jakewharton.rxbinding4.view.clicks
import com.jakewharton.rxbinding4.widget.textChanges
import com.jinyang.jetpackdemo.paging3.Repo
import com.jinyang.jetpackdemo.paging3.RepoList
import com.trello.rxlifecycle4.android.ActivityEvent
import com.trello.rxlifecycle4.components.support.RxAppCompatActivity
import okhttp3.Interceptor
import okhttp3.OkHttpClient
import retrofit2.Call
import retrofit2.Retrofit
import retrofit2.adapter.rxjava2.RxJava2CallAdapterFactory
import retrofit2.converter.gson.GsonConverterFactory
import retrofit2.http.*
import java.lang.IllegalArgumentException
import java.lang.NumberFormatException
import java.lang.RuntimeException
import android.R.string.no
import android.R.string.no
import android.R.string.no
import io.reactivex.rxjava3.functions.*
import io.reactivex.rxjava3.functions.Function
import android.R.string.no
import android.R.string.no
import androidx.lifecycle.lifecycleScope
import io.reactivex.rxjava3.observables.ConnectableObservable
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.launch
import java.io.IOException


class RxjavaDemoActivity : RxAppCompatActivity() {
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_rxjava_demo)
//        test1()
//        test2()
//        test3()
//        test4()
//        test5()
//        test6()
//        test7()
//        test8()
//        test9()
//        val btn3 = findViewById<Button>(R.id.btn_3)
//        btn3.clicks().subscribe {
//            test10()
//        }
////        test11()
//        test12()
//        test13()
        test14()
    }

    private fun test14() {
        LjyLogUtil.d("test14:" + Thread.currentThread())
        lifecycleScope.launch {
            LjyLogUtil.d("lifecycleScope.launch:" + Thread.currentThread())
            flow {
                LjyLogUtil.d("flow:" + Thread.currentThread())
                for (i in 1..3) {
                    emit(i)
                    Thread.sleep(500)
                }
                 throw RuntimeException("出错误了。。。")
                LjyLogUtil.d("flow: end")
            }
                .flowOn(Dispatchers.IO)
                .onStart {
                    LjyLogUtil.d("onStart:" + Thread.currentThread())
                }
                .flowOn(Dispatchers.Main)
                .onEach {
                    LjyLogUtil.d("onEach:${it}_" + Thread.currentThread())
                }
                .flowOn(Dispatchers.IO)
                .catch {
                    LjyLogUtil.d("catch:${it}_" + Thread.currentThread())
                }
                .flowOn(Dispatchers.Unconfined)
                .onCompletion {
                    LjyLogUtil.d("onCompletion:${it}_" + Thread.currentThread())
                }
                .flowOn(Dispatchers.Default)
                .collect {
                    LjyLogUtil.d("collect:${it}_" + Thread.currentThread())
                }
        }
    }

    private fun test13() {
        val interval: ConnectableObservable<Long> =
            Observable.interval(1, TimeUnit.SECONDS).publish()
        var disposable: Disposable? = null
        findViewById<Button>(R.id.btn_connect)
            .clicks().subscribe {
                disposable = interval.connect()
            }
        findViewById<Button>(R.id.btn_dispose)
            .clicks().subscribe {
                disposable?.dispose()
            }
        findViewById<Button>(R.id.btn_subscribe3)
            .clicks().subscribe {
                interval.subscribe {
                    LjyLogUtil.d("观察者3：$it")
                }
            }
        findViewById<Button>(R.id.btn_subscribe4)
            .clicks().subscribe {
                interval.subscribe {
                    LjyLogUtil.d("观察者4：$it")
                }
            }
    }

    private fun test12() {
        val interval = Observable.interval(1, TimeUnit.SECONDS)
        var disposable1: Disposable? = null
        val observer1 = object : Observer<Long> {
            override fun onSubscribe(d: Disposable) {
                disposable1 = d
            }

            override fun onNext(t: Long) {
                LjyLogUtil.d("观察者1：$t")
            }

            override fun onError(e: Throwable) {

            }

            override fun onComplete() {

            }
        }

        var disposable2: Disposable? = null
        val observer2 = object : Observer<Long> {
            override fun onSubscribe(d: Disposable) {
                disposable1 = d
            }

            override fun onNext(t: Long) {
                LjyLogUtil.d("观察者2：$t")
            }

            override fun onError(e: Throwable) {

            }

            override fun onComplete() {

            }
        }

        findViewById<Button>(R.id.btn_subscribe1)
            .clicks().subscribe {
                interval.subscribe(observer1)
            }
        findViewById<Button>(R.id.btn_dispose1)
            .clicks().subscribe {
                disposable1?.dispose()
            }


        findViewById<Button>(R.id.btn_subscribe2)
            .clicks().subscribe {
                interval.subscribe(observer2)
            }
        findViewById<Button>(R.id.btn_dispose2)
            .clicks().subscribe {
                disposable2?.dispose()
            }
    }

    private fun test11() {
        var result = "数据来自："
        val net = Observable.just("网络")
        val disk = Observable.just("磁盘")
        //使用merge，从网络和本地获取数据并展示
        Observable.merge(net, disk)
            .subscribe({
                result += "$it, "
            }, {

            }, {
                LjyLogUtil.d("result: $result")
            })
        //使用zip,合并2个网络请求向获取数据并展示
        val retrofit = Retrofit.Builder()
            .baseUrl("https://api.github.com/")
            .addConverterFactory(GsonConverterFactory.create())
            .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
            .build()
        val apiService = retrofit.create(ApiService::class.java)
        val repo1 = apiService.getItem(1001).subscribeOn(Schedulers.io())
        val repo2 = apiService.getItem(1002).subscribeOn(Schedulers.io())
        Observable.zip(
            repo1, repo2, { data1, data2 ->
                val repoList = ArrayList<RepoDetail>()
                repoList.add(data1)
                repoList.add(data2)
                repoList
            })
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe {
                for (repoDetail in it) {
                    LjyLogUtil.d("result: ${repoDetail.name}")
                }
            }

    }

    //内存，磁盘，网络 三级缓存
    var memoryCache: String? = null
    var diskCache: String? = null
    private fun test10() {
        val memory = Observable.create<String> {
            if (memoryCache != null) {
                it.onNext(memoryCache)
            } else {
                it.onComplete()
            }
        }

        val disk = Observable.create<String> {
            if (diskCache != null) {
                it.onNext(diskCache)
            } else {
                it.onComplete()
            }
        }.doOnNext {
            memoryCache = "内存数据"
        }

        val net = Observable.just("网络数据")
            .doOnNext {
                // memoryCache="内存数据"
                diskCache = "磁盘数据"
            }
        //通过concat（）合并memory、disk、network 3个被观察者的事件（即检查内存缓存、磁盘缓存 & 发送网络请求）
        Observable.concat(memory, disk, net)
            //通过firstElement()，从串联队列中取出并发送第1个有效事件（Next事件），即依次判断检查memory、disk、network
            .firstElement()
            .subscribe {
                LjyLogUtil.d("accept: $it")
            }
    }

    /**
    - SubscribeOn：指定被观察者Observable的工作线程
    - ObserveOn：指定观察者的observer工作线程
    - doOnEach：数据源（Observable）每发送一次数据，就调用一次
    - doOnNext：数据源每次调用onNext() 之前都会先回调该方法
    - doOnError：数据源每次调用onError() 之前会回调该方法。
    - doOnComplete：数据源每次调用onComplete() 之前会回调该方法
    - doOnSubscribe：数据源每次调用onSubscribe() 之后会回调该方法
    - doOnDispose：数据源每次调用dispose() 之后会回调该方法
     */
    private fun test9() {
        val observer = object : Observer<Int> {
            override fun onSubscribe(d: Disposable) {
                LjyLogUtil.d("${Thread.currentThread().name}_onSubscribe")
            }

            override fun onNext(t: Int) {
                LjyLogUtil.d("${Thread.currentThread().name}_onNext:$t")
            }

            override fun onError(e: Throwable) {
                LjyLogUtil.d("${Thread.currentThread().name}_onError:${e.message}")
            }

            override fun onComplete() {
                LjyLogUtil.d("${Thread.currentThread().name}_onComplete")
            }
        }
        Observable.just(1, 2, 3, 4)
            .doOnSubscribe { LjyLogUtil.d("${Thread.currentThread().name}_doOnSubscribe") }
            .doOnEach { LjyLogUtil.d("${Thread.currentThread().name}_doOnEach:$it") }
            .doOnNext { LjyLogUtil.d("${Thread.currentThread().name}_doOnNext:$it") }
            .doOnError { LjyLogUtil.d("${Thread.currentThread().name}_doOnError:${it.localizedMessage}") }
            .doOnComplete { LjyLogUtil.d("${Thread.currentThread().name}_doOnComplete") }
            .doOnDispose { LjyLogUtil.d("${Thread.currentThread().name}_doOnDispose") }
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(observer)
    }

    /**
     * 结合 RxBinding
     * RxBinding是对 Android View 事件的扩展, 它使得开发者可以对 View 事件使用 RxJava 的各种操作
     */
    private fun test8() {
        //按钮防抖
        val btn1 = findViewById<Button>(R.id.btn_1)
        btn1.clicks()
            .throttleFirst(2, TimeUnit.SECONDS)
            .subscribeOn(AndroidSchedulers.mainThread())
            .subscribe {
                LjyLogUtil.d("点击按钮")
            }
        //editText输入监听
        val et1 = findViewById<EditText>(R.id.et_1)
        et1.textChanges()
            .debounce(1, TimeUnit.SECONDS)
            //跳过第1次请求 因为初始输入框的空字符状态
            .skip(1)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe { it ->
                LjyLogUtil.d("${Thread.currentThread().name}_onSuccess:$it")
            }
        //联合/表单判断
        val etName = findViewById<EditText>(R.id.et_name)
        val etPwd = findViewById<EditText>(R.id.et_pwd)
        val obName = etName.textChanges()
        val obPwd = etPwd.textChanges()
        Observable.combineLatest(
            obName, obPwd, { name, pwd -> name == "ljy" && pwd == "123" })
            //跳过第1次请求 因为初始输入框的空字符状态
            .skip(1)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe { isLogin -> LjyLogUtil.d(if (isLogin) "登录成功" else "登录失败") }

        //定时器任务
        val time = 10L
        val btnLogin = findViewById<Button>(R.id.btn_login)
        btnLogin.clicks()
            .throttleFirst(time, TimeUnit.SECONDS)
            .subscribeOn(AndroidSchedulers.mainThread())
            .doOnNext { btnLogin.isEnabled = false }
            .subscribe {
                LjyLogUtil.d("点击登录")
                Observable.intervalRange(
                    0, time, 0, 1,
                    TimeUnit.SECONDS, AndroidSchedulers.mainThread()
                )
                    .subscribe(
                        { btnLogin.text = "剩余${time - it}秒" },
                        { LjyLogUtil.e(it.message) },
                        {
                            btnLogin.text = "获取验证码"
                            btnLogin.isEnabled = true
                        })
            }
        Observable.intervalRange(0, 60, 0, 1, TimeUnit.SECONDS)
            //利用RxLifecycle来解决内存泄漏问题
            //bindToLifecycle的自动取消订阅示例
            //.compose(bindToLifecycle())
            //手动设置在activity onPause的时候取消订阅
            .compose(this.bindUntilEvent(ActivityEvent.PAUSE))
            .subscribe(
                { LjyLogUtil.d("剩余${60 - it}秒") },
                { LjyLogUtil.e(it.message) },
                { LjyLogUtil.d("完成") }
            )
        //网络请求嵌套
        val retrofit = Retrofit.Builder()
            .baseUrl("https://api.github.com/")
            .addConverterFactory(GsonConverterFactory.create())
            .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
            .build()
        val apiService = retrofit.create(ApiService::class.java)
        apiService.searchRepo(emptyMap())
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .doOnNext {
                //展示列表
            }
            .flatMap {
                Observable.fromIterable(it.items).map { repo -> repo.id }
            }
            .observeOn(Schedulers.io())
            .flatMap {
                apiService.getItem(it)
            }
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe {
                //展示详情
            }
        Observable.interval(10, TimeUnit.SECONDS)
            .doOnNext {
                apiService.searchRepo(emptyMap())
                    .subscribe { LjyLogUtil.d("accept: ${it.items}") }
            }.subscribe { LjyLogUtil.d("第 $it 次轮询") }
        var count = 0
        apiService.searchRepo(emptyMap())
            .repeatWhen {
                it.flatMap {
                    if (count > 3) {
                        Observable.error(Throwable("轮询结束"))
                    } else {
                        LjyLogUtil.d("第 $count 次轮询")
                        Observable.just(1).delay(60, TimeUnit.SECONDS)
                    }
                }
            }.subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe({
                LjyLogUtil.d("accept: ${it.items}")
                count++
            }, {
                LjyLogUtil.d(it.message)
            })

        //最大重试次数
        val maxConnectCount = 10
        //已重试次数
        var currentRetryCount = 0
        apiService.searchRepo(emptyMap())
            .retryWhen {
                it.flatMap { throwable ->
                    //根据异常类型选择是否重试
                    if (throwable is IOException) {
                        if (currentRetryCount < maxConnectCount) {
                            currentRetryCount++
                            //遇到的异常越多，重试延迟间隔时间越长
                            Observable.just(1).delay(currentRetryCount * 60L, TimeUnit.SECONDS)
                        } else {
                            Observable.error(Throwable("重试结束"))
                        }
                    } else {
                        Observable.error(Throwable("发生了非网络异常（非I/O异常）"))
                    }
                }
            }
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe({
                LjyLogUtil.d("accept: ${it.items}")
            }, {
                LjyLogUtil.d(it.message)
            })

    }


    interface ApiService {
        @Headers("Content-Type: application/json;charset=UTF-8")
        @POST("api/search")
        fun searchRepo(@Body params: Map<String, Any>): Observable<RepoList>

        @GET("api/items/{userId}")
        fun getItem(@Path("userId") userId: Int): Observable<RepoDetail>
    }

    class RepoList {
        @SerializedName("items")
        val items: List<Repo> = emptyList()
    }

    data class Repo(
        @SerializedName("id") val id: Int,
        @SerializedName("name") val name: String,
        @SerializedName("description") val description: String,
        @SerializedName("stargazers_count") val starCount: String,
    )

    data class RepoDetail(
        @SerializedName("name")
        val name: String,
        @SerializedName("description")
        val description: String
    )

    /**
     * 过滤操作 Debounce, Distinct, ElementAt, Filter, First, IgnoreElements, Last, Sample, Skip, SkipLast, Take, TakeLast
     */
    private fun test7() {
        //观察者
        val observer = object : Observer<Int> {
            override fun onSubscribe(d: Disposable) {
                LjyLogUtil.d("${Thread.currentThread().name}_onSubscribe")
            }

            override fun onNext(t: Int) {
                LjyLogUtil.d("${Thread.currentThread().name}_onNext:$t")
            }

            override fun onError(e: Throwable) {
                LjyLogUtil.d("${Thread.currentThread().name}_onError:${e.message}")
            }

            override fun onComplete() {
                LjyLogUtil.d("${Thread.currentThread().name}_onComplete")
            }
        }

        //skip
        val observable = Observable.just(1, 2, 3, 4, 5, 6, 7, 8)
        observable.skipLast(4).subscribe(observer)
        observable.skip(4).subscribe(observer)
        //take
        observable.takeLast(4).subscribe(observer)
        observable.take(4).subscribe(observer)
        //debounce
        val observable2 = Observable.create<Int> {
            it.onNext(1)
            Thread.sleep(400)
            it.onNext(2)
            Thread.sleep(1200)
            it.onNext(3)
            Thread.sleep(1000)
            it.onNext(4)
            Thread.sleep(800)
            it.onNext(5)
            Thread.sleep(2000)
            it.onNext(6)
        }
        observable2
            .subscribeOn(Schedulers.io())
            .observeOn(Schedulers.io())
            .debounce(1, TimeUnit.SECONDS)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(observer)
        //sample:与debounce的区别是，sample是以时间为周期的发射，一秒又一秒内的最新数据。而debounce是最后一个有效数据开始
        observable2
            .subscribeOn(Schedulers.io())
            .observeOn(Schedulers.io())
            .sample(1, TimeUnit.SECONDS)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(observer)
        //throttleFirst & throttleLast & throttleWithTimeout & throttleLatest
        //hrottleFirst是指定周期内第一个数据，throttleLast与smaple一致。throttleWithTimeout与debounce一致。
        observable2
            .subscribeOn(Schedulers.io())
            .observeOn(Schedulers.io())
            .throttleFirst(1, TimeUnit.SECONDS)
            .throttleLast(1, TimeUnit.SECONDS)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(observer)
        //timeout: 后一个数据发射未在前一个元素发射后规定时间内发射则返回超时异常。
        observable2
            .subscribeOn(Schedulers.io())
            .observeOn(Schedulers.io())
            .timeout(1, TimeUnit.SECONDS)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(observer)
        //distinct
        Observable.just(1, 2, 3, 4, 3, 2, 1)
            .distinct()
//            .distinctUntilChanged()//去掉相邻重复数据
            .subscribe(observer)
        //elementAt
        val maybeObserver = object : MaybeObserver<Int> {
            override fun onSubscribe(d: Disposable) {
                LjyLogUtil.d("${Thread.currentThread().name}_onSubscribe")
            }

            override fun onSuccess(t: Int) {
                LjyLogUtil.d("${Thread.currentThread().name}_onSuccess:$t")
            }

            override fun onError(e: Throwable) {
                LjyLogUtil.d("${Thread.currentThread().name}_onError:${e.message}")
            }

            override fun onComplete() {
                LjyLogUtil.d("${Thread.currentThread().name}_onComplete")
            }
        }
        Observable.just(4, 3, 2, 1)
            .elementAt(1)
            .subscribe(maybeObserver)
        //filter
        Observable.just(1, 2, 3, 4, 5)
            .filter {
                it % 2 == 0
            }.subscribe(observer)
        //first
        Observable.just(1, 2, 3, 4, 5)
//            .first(-1)
//            .firstOrError()
            .firstElement()
            .subscribe(maybeObserver)
        //last
        Observable.just(1, 2, 3, 4, 5)
//            .last(-1)
//            .lastOrError()
            .lastElement()
            .subscribe(maybeObserver)
        //ignoreElements 作用于Flowable、Observable。ignoreElement作用于Maybe、Single
        Observable.just(1, 2, 3, 4, 5)
            .ignoreElements()
            .subscribe(object : CompletableObserver {
                override fun onSubscribe(d: Disposable) {
                    LjyLogUtil.d("${Thread.currentThread().name}_onSubscribe")
                }

                override fun onComplete() {
                    LjyLogUtil.d("${Thread.currentThread().name}_onComplete")
                }

                override fun onError(e: Throwable) {
                    LjyLogUtil.d("${Thread.currentThread().name}_onError:${e.message}")
                }
            })
        //ofType作用于Flowable、Observable、Maybe，过滤选择类型
        Observable.just(1, 2.1, "3", 4L, 5.0)
            .ofType(Int::class.java)
            .subscribe(observer)
        val observer3 = object : Observer<Serializable> {
            override fun onSubscribe(d: Disposable) {
                LjyLogUtil.d("${Thread.currentThread().name}_onSubscribe")
            }

            override fun onNext(t: Serializable) {
                LjyLogUtil.d("${Thread.currentThread().name}_onNext:$t")
            }

            override fun onError(e: Throwable) {
                LjyLogUtil.d("${Thread.currentThread().name}_onError:${e.message}")
            }

            override fun onComplete() {
                LjyLogUtil.d("${Thread.currentThread().name}_onComplete")
            }
        }
        val just1 = Observable.just(1, 2, 3)
        val just2 = Observable.just("A", "B", "C")
        val just3 = Observable.just(4, 5, 6)
        //merge和concat的区别:merge():合并后发射的数据项是无序的，concat():合并后发射的数据项是有序的。
        Observable.merge(just1, just2).subscribe(observer3)
        Observable.concat(just1, just2).subscribe(observer3)
        Observable.concatArray(
            Observable.just(1, 2, 3),
            Observable.just(4, 5, 6),
            Observable.just(7, 8, 9)
        ).subscribe(observer3)
        Observable.just(1, 2, 3, 4, 5, 6)
            .collect(
                { ArrayList() },
                BiConsumer<ArrayList<Int?>, Int> { t1, t2 -> t1.add(t2) }
            ).subscribe(Consumer { LjyLogUtil.d("num: $it") })
        //mergeWith
        just1.mergeWith(just3).subscribe(observer3)
        //zip操作符是将两个数据流进行指定的函数规则合并
        Observable.zip(just1, just2, { t1, t2 -> "${t1}_${t2}" })
            .subscribe(observer3)
        //zipWith
        just1.zipWith(just2, { t1, t2 -> "${t1}_${t2}" })
            .subscribe(observer3)
        //startWith操作符是将另一个数据流合并到原数据流的开头
        val just4 = Observable.just("1", "2", "3")
        just4.startWith(just2)
            .subscribe(observer3)
        Observable.just(1, 2, 3)
            .startWith(Observable.just(4, 5, 6))
            .startWithArray(7, 8, 9)
            .subscribe(observer3)
        Observable.just(1, -2, 3)
            .all {
                it > 0
            }.subscribe { it ->
                LjyLogUtil.d("${Thread.currentThread().name}_onSuccess:$it")
            }
        Observable.just(1, -2, 3)
            .contains(2)
            .subscribe { it ->
                LjyLogUtil.d("${Thread.currentThread().name}_onSuccess:$it")
            }
        Observable.just(1, -2, 3)
            .isEmpty()
            .subscribe { it ->
                LjyLogUtil.d("${Thread.currentThread().name}_onSuccess:$it")
            }
        Observable.empty<Int>()
            .defaultIfEmpty(-1)
            .subscribe { it ->
                LjyLogUtil.d("${Thread.currentThread().name}_onSuccess:$it")
            }
        val list: MutableList<ObservableSource<Int>> = ArrayList()
        list.add(Observable.just(1, 2, 3).delay(1, TimeUnit.SECONDS))
        list.add(Observable.just(4, 5, 6))
        Observable.amb(list).subscribe {
            LjyLogUtil.d("${Thread.currentThread().name}_onSuccess:$it")
        }
        // 从0开始每1s发送1个数据
        Observable.interval(1, TimeUnit.SECONDS)
            .takeWhile {
                it < 10
            }
        Observable.interval(1, TimeUnit.SECONDS)
            .takeUntil {
                it > 10
            }
        Observable.interval(1, TimeUnit.SECONDS)
            .skipUntil(Observable.timer(5, TimeUnit.SECONDS))
        Observable.sequenceEqual(
            Observable.just(4, 5, 6),
            Observable.just(4, 5, 6)
        ).subscribe { it ->
            LjyLogUtil.d("${Thread.currentThread().name}_onSuccess:$it")
        }
        Observable.interval(1, TimeUnit.SECONDS)
            .skipWhile {
                it < 10
            }
        Observable.just(1, 2, 3)
            .count()
            .subscribe(object : SingleObserver<Long> {
                override fun onSubscribe(d: Disposable) {
                    LjyLogUtil.d("${Thread.currentThread().name}_onSubscribe")
                }

                override fun onSuccess(t: Long) {
                    LjyLogUtil.d("${Thread.currentThread().name}_onSuccess:$t")
                }

                override fun onError(e: Throwable) {
                    LjyLogUtil.d("${Thread.currentThread().name}_onError:${e.message}")
                }

            })
        //join操作符是有时间期限的合并操作符
        just1.join(just2,
            //规定just2的过期期限
            { Observable.timer(3, TimeUnit.SECONDS) },
            //规定just1的过期期限
            { Observable.timer(8, TimeUnit.SECONDS) },
            //规定just1和just2的合并规则
            { t1, t2 -> "${t1}_${t2}" })
            .subscribe(observer3)
        //combineLatest:在结合不同数据源时，发射速度快的数据源最新item与较慢的相结合
        Observable.combineLatest(just1, just2,
            { t1, t2 -> "${t1}_${t2}" }).subscribe(observer3)
        //switchOnNext:一个发射多个小数据源的数据源，这些小数据源发射数据的时间发生重复时，取最新的数据源
        Observable.switchOnNext(ObservableSource<Observable<Int>> { }, 12)

        //map
        Observable.just("1", "2", "3").map { it.toInt() }.subscribe(observer)
        //concatMap与flatMap的区别:    concatMap是有序的，flatMap是无序的
        Observable.just("A", "B", "C")
            .flatMap { m ->
                Observable.intervalRange(1, 3, 0, 1, TimeUnit.SECONDS).map { n ->
                    "($m,$n)"
                }
            }
        Observable.just("A", "B", "C")
            .concatMap { m ->
                Observable.intervalRange(1, 3, 0, 1, TimeUnit.SECONDS).map { n ->
                    "($m,$n)"
                }
            }

        val observer4 = object : Observer<List<String>> {
            override fun onSubscribe(d: Disposable) {
                LjyLogUtil.d("${Thread.currentThread().name}_onSubscribe")
            }

            override fun onNext(t: List<String>) {
                LjyLogUtil.d("${Thread.currentThread().name}_onNext:$t")
            }

            override fun onError(e: Throwable) {
                LjyLogUtil.d("${Thread.currentThread().name}_onError:${e.message}")
            }

            override fun onComplete() {
                LjyLogUtil.d("${Thread.currentThread().name}_onComplete")
            }
        }
        //groupBy操作符可以将发射出来的数据项进行分组
        Observable.just(
            "Tiger",
            "Elephant",
            "Cat",
            "Chameleon",
            "Frog",
            "Fish",
            "Turtle",
            "Flamingo"
        )
            .groupBy { it[0].uppercaseChar() }
            .concatMapSingle { it.toList() }
            .subscribe(observer4)
        //scan操作符会对发射的数据和上一轮发射的数据进行函数处理，并返回的数据供下一轮使用，持续这个过程来产生剩余的数据流。其应用场景有简单的累加计算，判断所有数据的最小值等
        Observable.just(1, 2, 3, 4)
            .scan { t1, t2 -> t1 + t2 }
            .subscribe(observer)
        //buffer操作符可以将发射出来的数据流，在给定的缓存池中进行缓存
        val observer5 = object : Observer<List<Int>> {
            override fun onSubscribe(d: Disposable) {
                LjyLogUtil.d("${Thread.currentThread().name}_onSubscribe")
            }

            override fun onNext(t: List<Int>) {
                LjyLogUtil.d("${Thread.currentThread().name}_onNext:$t")
            }

            override fun onError(e: Throwable) {
                LjyLogUtil.d("${Thread.currentThread().name}_onError:${e.message}")
            }

            override fun onComplete() {
                LjyLogUtil.d("${Thread.currentThread().name}_onComplete")
            }
        }
        Observable.just(1, 2, 3, 4, 5, 6, 7, 8)
            .buffer(3)
            .subscribe(observer5)
        //window操作符和buffer操作符在功能上实现的效果是一样的，但window操作符最大区别在于同样是缓存一定数量的数据项，
        // window操作符最终发射出来的是新的事件流integerObservable，而buffer操作符发射出来的是新的数据流，
        // 也就是说，window操作符发射出来新的事件流中的数据项，还可以经过Rxjava其他操作符进行处理。
        val observer6 = object : Observer<Observable<Int>> {
            override fun onSubscribe(d: Disposable) {
                LjyLogUtil.d("${Thread.currentThread().name}_onSubscribe")
            }

            override fun onNext(t: Observable<Int>) {
                LjyLogUtil.d("${Thread.currentThread().name}_onNext:$t")
            }

            override fun onError(e: Throwable) {
                LjyLogUtil.d("${Thread.currentThread().name}_onError:${e.message}")
            }

            override fun onComplete() {
                LjyLogUtil.d("${Thread.currentThread().name}_onComplete")
            }
        }
        Observable.just(1, 2, 3, 4)
            .window(2, 1)
            .subscribe(observer6)
        //cast: 将数据元素转型成其他类型,转型失败会抛出异常
        Observable.just(1, 4.0, 3f, 7, 12, 4.6, 5)
            .cast(Int::class.java)
            .subscribe(observer)
        //onErrorReturn:调用数据源的onError函数后会回到该函数，可对错误进行处理，然后返回值，会调用观察者onNext()继续执行，执行完调用onComplete()函数结束所有事件的发射。
        Observable.just(1, 2, 3.2, 4)
            .map { it.toInt() }
            .onErrorReturn {
                if (it is NumberFormatException) {
                    0
                } else {
                    throw IllegalArgumentException()
                }
            }.subscribe(observer)
        //与onErrorReturn类似，onErrorReturnItem不对错误进行处理，直接返回一个值
        Observable.just(1, 2, 3.2, 4)
            .map { it.toInt() }
            .onErrorReturnItem(0)
            .subscribe(observer)
        //onErrorResumeNext: 遇到错误时，发送1个新的Observable
        Observable.just(1, 2, 3.2, 4)
            .map { it.toInt() }
            .onErrorResumeNext { Observable.just(5, 6, 7) }
            .subscribe(observer)
        //onExceptionResumeNext:遇到错误时，发送1个新的Observable
        Observable.just(1, 2, 3.2, 4)
            .map { it.toInt() }
            .retryWhen {
                it.flatMap { throwable ->
                    if (throwable is NumberFormatException) {
                        Observable.error(Throwable("retryWhen终止"))
                    } else {
                        Observable.just(5, 6, 7)
                    }
                }
            }
            .subscribe(observer)
        //retry:当发生错误时，数据源重复发射item，直到没有异常或者达到所指定的次数
        Observable.just(1, 2, 3, 4)
            .retry(3)
            .subscribe(observer)
        //发生异常时，返回值是false表示继续执行(重复发射数据)，true不再执行,但会调用onError方法。
        var temp = 0
        Observable.just(1, 2, 3, 4)
            .map {
                temp = it
                it
            }
            .retryUntil {
                temp > 3
            }
            .subscribe(observer)
    }

    /**
     * 创建操作 Create, Defer, Empty/Never/Throw, From, Interval, Just, Range, Repeat, Start, Timer
     */
    private fun test6() {
        //观察者
        val observer = object : Observer<Int> {
            override fun onSubscribe(d: Disposable) {
                LjyLogUtil.d("${Thread.currentThread().name}_onSubscribe")
            }

            override fun onNext(t: Int) {
                LjyLogUtil.d("${Thread.currentThread().name}_onNext:$t")
            }

            override fun onError(e: Throwable) {
                LjyLogUtil.d("${Thread.currentThread().name}_onError:${e.message}")
            }

            override fun onComplete() {
                LjyLogUtil.d("${Thread.currentThread().name}_onComplete")
            }
        }
        //Create
        val observable1 = Observable.create(ObservableOnSubscribe<Int> {
            it.onNext(1)
            it.onNext(3)
            it.onNext(5)
            it.onComplete()
            LjyLogUtil.d("${Thread.currentThread().name}_subscribe")
        })
        observable1.subscribe(observer)
        //from
        val numList = listOf(1, 2, 3, 4)
        val observable21 = Observable.fromArray(arrayOf(1, 2, 3))
        val observable22 = Observable.fromIterable(listOf(4, 5, 6))
        observable21.subscribe { TODO("Not yet implemented") }
        observable22.subscribe(observer)
        Observable.never<Int>().subscribe(observer)
        Observable.empty<Int>().subscribe(observer)
        Observable.error<Int>(RuntimeException()).subscribe(observer)
        //defer
        var num = 1
        val observable = Observable.defer { Observable.just(num) }
        num = 2
        observable.subscribe(object : Observer<Int> {
            override fun onSubscribe(d: Disposable) {
                LjyLogUtil.d("${Thread.currentThread().name}_onSubscribe")
            }

            override fun onNext(t: Int) {
                LjyLogUtil.d("${Thread.currentThread().name}_onNext:$t")
            }

            override fun onError(e: Throwable) {
                LjyLogUtil.d("${Thread.currentThread().name}_onError:${e.message}")
            }

            override fun onComplete() {
                LjyLogUtil.d("${Thread.currentThread().name}_onComplete")
            }
        })

        //just
        val observable3 = Observable.just(1, 2, 3, 4)
        observable3.subscribe(observer)
        Observable.just(1, 2, 3)
            .delay(1, TimeUnit.SECONDS)
            .subscribe(observer)
        //range
        val observable4 = Observable.range(0, 5)
        observable4.subscribe(observer)
        //interval
        //初始延时1秒，每3秒发一个自增整数
        Observable.interval(1, 3, TimeUnit.SECONDS)
        //初始延时2秒，后每1秒发一个从10开始的整数，发5个（发到14）停止
        val observable5 = Observable.intervalRange(10, 5, 2, 1, TimeUnit.SECONDS);
        observable5.map { it.toInt() }.subscribe(observer)
        //repeat
        //一直重复
        Observable.fromArray(1, 2, 3, 4).repeat();
        //重复发送5次
        Observable.fromArray(1, 2, 3, 4).repeat(5);
        //创建一个重复发射指定数据或数据序列的Observable，它依赖于另一个Observable发射的数据
        Observable.just(1, 2, 4)
            .repeatWhen {
                it.flatMap { obj ->
                    if (obj is NumberFormatException) {
                        Observable.error(Throwable("retryWhen终止"))
                    } else {
                        Observable.just(5, 6, 7)
                    }
                }
            }.subscribe(observer)
        //重复发送直到符合条件时停止重复
        val observable6 = Observable.fromArray(1, 2, 3, 4).repeatUntil { false };
        observable6.subscribe(observer)
        //timer
        //延时3秒后，发送一个整数0
        val observable7 = Observable.timer(3, TimeUnit.SECONDS)
        observable7.map { it.toInt() }.subscribe(observer)

    }

    /**
     * Maybe/MaybeObserver
     * 如果可能发送一个数据或者不会发送任何数据，这时候你就需要Maybe，它类似于Single和Completable的混合体。
     * onSuccess和onComplete是互斥的存在
     */
    private fun test5() {
        Maybe.create(MaybeOnSubscribe<String> {
            it.onSuccess("str1")
            it.onComplete()
        }).subscribe(object : MaybeObserver<String> {
            override fun onSubscribe(d: Disposable) {
                LjyLogUtil.d("${Thread.currentThread().name}_onSubscribe")
            }

            override fun onSuccess(t: String) {
                LjyLogUtil.d("${Thread.currentThread().name}_onSuccess:$t")
            }

            override fun onError(e: Throwable) {
                LjyLogUtil.d("${Thread.currentThread().name}_onError:${e.message}")
            }

            override fun onComplete() {
                LjyLogUtil.d("${Thread.currentThread().name}_onComplete")
            }

        })
    }

    /**
     * Completable/CompletableObserver
     * 如果你的观察者连onNext事件都不关心，可以使用Completable，它只有onComplete和onError两个事件
     * 要转换成其他类型的被观察者，也是可以使用toFlowable()、toObservable()等方法去转换。
     */
    private fun test4() {
        Completable.create { emitter ->
            LjyLogUtil.d("${Thread.currentThread().name}_subscribe")
            emitter.onComplete()//单一onComplete或者onError
        }.subscribe(object : CompletableObserver {
            override fun onSubscribe(d: Disposable) {
                LjyLogUtil.d("${Thread.currentThread().name}_onSubscribe")
            }

            override fun onComplete() {
                LjyLogUtil.d("${Thread.currentThread().name}_onComplete")
            }

            override fun onError(e: Throwable) {
                LjyLogUtil.d("${Thread.currentThread().name}_onError:${e.message}")
            }

        })
    }

    /**
     * Single/SingleObserver
     * 它总是只发射一个值，或者一个错误通知，而不是发射一系列的值（当然就不存在背压问题）
     */
    private fun test3() {
        Single.just(1)
            .map { "num_$it" }
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe { num ->
                LjyLogUtil.d(num)
            }
        Single.create(SingleOnSubscribe<String> { emitter ->
            LjyLogUtil.d("${Thread.currentThread().name}_subscribe")
            emitter.onSuccess("str1")
            emitter.onSuccess("str2")//错误写法，重复调用也不会处理，因为只会调用一次
        }).subscribe(object : SingleObserver<String> {
            override fun onSubscribe(d: Disposable) {
                LjyLogUtil.d("${Thread.currentThread().name}_onSubscribe")
            }

            override fun onSuccess(t: String) {
                LjyLogUtil.d("${Thread.currentThread().name}_onSuccess:$t")
            }

            override fun onError(e: Throwable) {
                LjyLogUtil.d("${Thread.currentThread().name}_onError:${e.message}")
            }

        })
    }

    /**
     *  Flowable/Subscriber
     * 上游的被观察者会响应下游观察者的数据请求，下游调用request(n)来告诉上游发送多少个数据。这样避免了大量数据堆积在调用链上，使内存一直处于较低水平
     */
    private fun test2() {
        var sub: Subscription? = null
        Flowable.create(FlowableOnSubscribe<Int> {
            it.onNext(1)
            it.onNext(3)
            it.onNext(5)
            it.onComplete()
            LjyLogUtil.d("${Thread.currentThread().name}_subscribe")
            //使用create创建Flowable，需要指定背压策略
        }, BackpressureStrategy.BUFFER)
        //BackpressureStrategy.ERROR:缓存池满了，就会抛出异常
        //BackpressureStrategy.BUFFER:缓存池满了，就会等待下游处理
        //BackpressureStrategy.DROP:缓存池满了，就会丢弃多余事件
        //BackpressureStrategy.LATEST:缓存池，只会存储128个事件
        Flowable.rangeLong(0, 5)
        Flowable.range(0, 5)
            .subscribe(object : Subscriber<Int> {
                override fun onSubscribe(s: Subscription?) {
                    //当订阅后，会首先调用这个方法，其实就相当于onStart()，
                    //传入的Subscription s参数可以用于请求数据或者取消订阅
                    LjyLogUtil.d("${Thread.currentThread().name}_onSubscribe start")
                    sub = s
                    sub?.request(1)
                    LjyLogUtil.d("${Thread.currentThread().name}_onSubscribe end")
                }

                override fun onNext(t: Int?) {
                    LjyLogUtil.d("${Thread.currentThread().name}_onNext:$t")
                    sub?.request(1)
                }

                override fun onError(t: Throwable?) {
                    LjyLogUtil.d("${Thread.currentThread().name}_onError:${t?.message}")
                }

                override fun onComplete() {
                    LjyLogUtil.d("${Thread.currentThread().name}_onComplete")
                }

            })

    }

    /**
     *  Observable/Observer
     *  不支持背压：当被观察者快速发送大量数据时，下游不会做其他处理，即使数据大量堆积，调用链也不会报MissingBackpressureException,消耗内存过大只会OOM。
     *  （官方给出以1000个事件为分界线作为参考）
     */
    private fun test1() {
        val observable = Observable.create(ObservableOnSubscribe<Int> {
            it.onNext(1)
            it.onNext(3)
            it.onNext(5)
            it.onComplete()
            LjyLogUtil.d("${Thread.currentThread().name}_subscribe")
        })
        //Disposable通过 dispose() ⽅方法来让上游停⽌止⼯工作，达到「丢弃」的效果

        var disposable: Disposable? = null
        val observer = object : Observer<String> {
            override fun onSubscribe(d: Disposable) {
                //订阅后发送数据之前, 回调这个方法，Disposable可用于取消订阅
                disposable = d
                LjyLogUtil.d("${Thread.currentThread().name}_onSubscribe")
            }

            override fun onNext(t: String) {
                LjyLogUtil.d("${Thread.currentThread().name}_onNext:$t")
            }

            override fun onError(e: Throwable) {
                LjyLogUtil.d("${Thread.currentThread().name}_onError:${e.message}")
            }

            override fun onComplete() {
                LjyLogUtil.d("${Thread.currentThread().name}_onComplete")
            }
        }

        //rxJava的整体结构是一条链，链的最上游是被观察者observable，最下游是观察者observer，
        //链的中间各个节点，既是其下游的observable，又是其上游的observer
        observable
            //subscribeOn: 切换起源 Observable 的线程,
            // 当多次调⽤用 subscribeOn() 的时候，只有最上⾯面的会对起源 Observable 起作⽤用，
            // 原因subscribeOn底层通过新建observable实现
            .subscribeOn(Schedulers.io())
            .subscribeOn(AndroidSchedulers.mainThread())
            //observeOn指定的是它之后的操作所在的线程，通过observeOn的多次调用，程序实现了线程的多次切换
            //影响范围observeOn是它下面的每个observer，除非又遇到新的observeOn
            .observeOn(Schedulers.io())
            .map {
                LjyLogUtil.d("map：${Thread.currentThread().name}")
                "num_$it"
            }
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(observer)

//        disposable?.dispose()
        var compositeDisposable: CompositeDisposable = CompositeDisposable()
        compositeDisposable.add(disposable)//增加某个事件
        compositeDisposable.addAll(disposable)//增加所有事件
        compositeDisposable.dispose()//释放所有事件
        compositeDisposable.clear()//释放所有事件，实现同dispose()
        compositeDisposable.remove(disposable)//移除某个事件并释放
        compositeDisposable.delete(disposable)//移除某个事件


    }
}